package com.atguigu.day08

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

class MyReceiver(hostname:String,port:Int) extends Receiver[String](StorageLevel.MEMORY_ONLY){
  //receiver启动时调用
  override def onStart(): Unit = {

    new Thread(){
      override def run(): Unit = receive()
    }.start()

  }
  //采集数据
  def receive(): Unit ={
    val socket = new Socket(hostname,port)

    val br = new BufferedReader(new InputStreamReader(socket.getInputStream))

    var line = br.readLine()

    while(line!=null){
      //存储数据,等到一个批次之后统一处理
      store(line)
      line = br.readLine()
    }

    br.close()

    socket.close()
  }
  //receiver停止时调用
  override def onStop(): Unit = {

  }
}
